﻿using Newtonsoft.Json;
using PLM.ConfigurationCenter.EventBus.Eventbus;
using Polly;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Exceptions;
using System.Net.Sockets;
using System.Text;

namespace PLM.ConfigurationCenter.EventBus.RabbitMQ {

    public class EventBusRabbitMQ : IDisposable {

        const string BROKER_NAME = "mapleleaf_event_bus";

        /// <summary>
        /// RabbitMQ 连接
        /// </summary>
        private readonly IRabbitMQConnection _rabbitMQConnection;
        /// <summary>
        /// 重试次数
        /// </summary>
        private readonly int _retryCount;

        private readonly IModel _consumerChannel;


        /// <summary>
        /// 构造函数
        /// </summary>
        /// <param name="mQConnection"></param>
        public EventBusRabbitMQ(IRabbitMQConnection mQConnection) {
            _rabbitMQConnection = mQConnection;
            _consumerChannel = CreateConsumerChannel();
        }


        /// <summary>
        /// 发布
        /// </summary>
        /// <param name="event">事件模型</param>
        public void Publish(IntegrationEvent @event, string mes) {
            if (!_rabbitMQConnection.IsConnected) {
                _rabbitMQConnection.TryConnect();
            }
            var policy = Policy.Handle<BrokerUnreachableException>()
                .Or<SocketException>()
                .WaitAndRetry(_retryCount, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)), (ex, time) => {
                    //_logger.LogWarning(ex, "Could not publish event: {EventId} after {Timeout}s ({ExceptionMessage})", @event.Id, $"{time.TotalSeconds:n1}", ex.Message);
                });

            var eventName = @event.GetType().Name;

            //声明一个交换器
            _consumerChannel.ExchangeDeclare(exchange: BROKER_NAME, ExchangeType.Direct);
            //创建一个消息队列
            _consumerChannel.QueueDeclare("PublishSubscrib01", true, false, false, null);
            //绑定交换器和消息队列
            _consumerChannel.QueueBind("PublishSubscrib01", BROKER_NAME, eventName, null);

            var message = JsonConvert.SerializeObject(mes);
            var body = Encoding.UTF8.GetBytes(message);
            policy.Execute(() => {
                var properties = _consumerChannel.CreateBasicProperties();
                properties.DeliveryMode = 2; // persistent
                _consumerChannel.BasicPublish(
                    exchange: BROKER_NAME,
                    routingKey: eventName,
                    mandatory: true,
                    basicProperties: properties,
                    body: body
                    );
            });
        }

        /// <summary>
        /// 订阅
        /// </summary>
        /// <typeparam name="T">约束：事件模型</typeparam>
        public void Subscribe(string queueName, Action<string> callback) {
            var consumer = new AsyncEventingBasicConsumer(_consumerChannel);
            consumer.Received += async (model, ea) => {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);
                await Task.Yield();
                callback?.Invoke(message);
                _consumerChannel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
            };
            _consumerChannel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
        }
        /// <summary>
        /// 创建消费通道
        /// </summary>
        /// <returns></returns>
        private IModel CreateConsumerChannel() {
            if (!_rabbitMQConnection.IsConnected) {
                _rabbitMQConnection.TryConnect();
            }
            return _rabbitMQConnection.CreateModel();
        }
        /// <summary>
        /// 
        /// </summary>
        public void Dispose() {
            _consumerChannel?.Dispose();
        }
    }
}
